package com.zhuiwei.bigdaa.hbase.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.ArrayList;

/**
 * @author zhuiwei
 * @version 2019
 */
public class HbaseService {
    private Logger log = LoggerFactory.getLogger(HbaseService.class);

    //申明静态配置
    private Configuration conf;
    private Connection connection = null;

    public HbaseService(Configuration conf) {
        this.conf = conf;
        try {
            connection = ConnectionFactory.createConnection(conf);
        } catch (IOException e) {
            log.error("获取HBase连接失败");
        }
    }

    /**
     * 判断表是否存在
     * @param tableName 表名
     * @return boolean
     */
    public boolean isTableExist(String tableName) throws Exception {
        HBaseAdmin admin=(HBaseAdmin)connection.getAdmin();
        return admin.tableExists(TableName.valueOf(tableName));
    }

    /**
     * 创建表
     * @param tableName 表名
     * @param constraint 存在是否删除
     * @param family 列族
     * @throws Exception 异常
     */
    public void createTable(String tableName,boolean constraint,String... family) throws Exception {
        TableName name=TableName.valueOf(tableName);
        HBaseAdmin admin =(HBaseAdmin) connection.getAdmin();
        if (admin.tableExists(name)){
            if (constraint){
                admin.disableTable(name);
                admin.deleteTable(name);
                HTableDescriptor descriptor =new HTableDescriptor(TableName.valueOf(tableName));
                for (String cf:family){
                    descriptor.addFamily(new HColumnDescriptor(cf));
                }
                admin.createTable(descriptor);
                log.info("创建表"+tableName+"成功");
            }else {
                log.error("表"+tableName+"已存在");
            }
        }else {
            HTableDescriptor descriptor =new HTableDescriptor(TableName.valueOf(tableName));
            for (String cf:family){
                descriptor.addFamily(new HColumnDescriptor(cf));
            }
            admin.createTable(descriptor);
            log.info("创建表"+tableName+"成功");
        }
    }

    /**
     * 向表中插入一条数据
     * @param tableName 表名
     * @param rowKey 行键
     * @param family 列族
     * @param qualifier 列
     * @param value 值
     * @throws Exception 异常
     */
    public  void addRowData(String tableName,String rowKey,String family,String qualifier,String value)throws Exception{
        TableName name = TableName.valueOf(tableName);
        HTable hTable = (HTable)connection.getTable(name);
        Put put=new Put(Bytes.toBytes(rowKey));
        put.addColumn(Bytes.toBytes(family),Bytes.toBytes(qualifier),Bytes.toBytes(value));
        hTable.put(put);
        hTable.close();
        log.info("插入成功");
    }

    /**
     * 获取表所有数据
     * @param tableName 表名
     * @throws Exception 异常
     */
    public String getAllRows(String tableName)throws Exception{
        StringBuilder data = new StringBuilder();
        TableName name = TableName.valueOf(tableName);
        HTable hTable = (HTable)connection.getTable(name);
        Scan scan = new Scan();
        ResultScanner resultScanner = hTable.getScanner(scan);
        for (Result result:resultScanner){
            Cell[] cells = result.rawCells();
            for (Cell cell:cells){
                String str=Bytes.toString(CellUtil.cloneRow(cell))
                        +"\t"
                        +Bytes.toString(CellUtil.cloneFamily(cell))
                        +"\t"
                        +Bytes.toString(CellUtil.cloneQualifier(cell))
                        +"\t"
                        +Bytes.toString(CellUtil.cloneValue(cell))
                        +"\n";
                data.append(str);
            }
        }
        return data.toString();
    }

    /**
     * 获取表中某一行所有数据
     * @param tableName 表名
     * @param rowKey 行键
     * @throws Exception 异常
     */
    public String getRow(String tableName,String rowKey)throws Exception{
        StringBuilder data = new StringBuilder();
        TableName name = TableName.valueOf(tableName);
        HTable hTable = (HTable)connection.getTable(name);
        Get get = new Get(Bytes.toBytes(rowKey));
        Result result = hTable.get(get);
        for (Cell cell:result.rawCells()){
            String str=Bytes.toString(CellUtil.cloneRow(cell))
                    +"\t"
                    +Bytes.toString(CellUtil.cloneFamily(cell))
                    +"\t"
                    +Bytes.toString(CellUtil.cloneQualifier(cell))
                    +"\t"
                    +Bytes.toString(CellUtil.cloneValue(cell))
                    +"\n";
            data.append(str);
        }
        return data.toString();
    }

    /**
     * 获取表中某一行指定"列族：列"的数据
     * @param tableName 表名
     * @param rowKey 行键
     * @param family 列族
     * @param qualifier 列
     * @throws Exception 异常
     */
    public String getRowQualifier(String tableName,String rowKey,String family,String qualifier)throws Exception{
        TableName name = TableName.valueOf(tableName);
        HTable hTable = (HTable)connection.getTable(name);
        Get get = new Get(Bytes.toBytes(rowKey));
        get.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier));
        Result result = hTable.get(get);
        return Bytes.toString(result.value());
    }

    /**
     * 删除多行数据
     * @param tableName 表名
     * @param rowKeys 行键
     * @throws Exception 异常
     */
    public void deleteMultiRow(String tableName,String...rowKeys)throws Exception{
        TableName name = TableName.valueOf(tableName);
        HTable hTable = (HTable)connection.getTable(name);
        ArrayList<Delete> deleteList = new ArrayList<>();
        for (String rowKey:rowKeys){
            Delete delete = new Delete(Bytes.toBytes(rowKey));
            deleteList.add(delete);
        }
        hTable.delete(deleteList);
        hTable.close();
        log.info("删除成功");
    }

    /**
     * 删除表
     * @param tableName 表名
     * @throws Exception 异常
     */
    public void drop(String tableName) throws Exception{
        HBaseAdmin admin=(HBaseAdmin)connection.getAdmin();
        if (admin.tableExists(tableName)){
            admin.disableTable(tableName);
            admin.deleteTable(tableName);
            System.out.println("表"+tableName+"删除成功");
        }else {
            System.out.println("表"+tableName+"不存在");
        }
    }
}
